import json
from flask import Flask, render_template
from flask_socketio import SocketIO
from kafka import KafkaConsumer


app = Flask(__name__)
app.config['SECRET_KEY'] = 'secret!'
socketio = SocketIO(app)
thread = None

#***************************************************
consumer1 = KafkaConsumer(
    auto_offset_reset='earliest',
    bootstrap_servers = "cheng:9092",
    group_id = "niit2",
    )
consumer1.subscribe(["result"])
#****************************************************

#**********************************************************
def background_thread2():
    inl = 0
    outl = 0

    for msg in consumer1:
        data_json = msg.value.decode('utf8')
        print('-----当前数据----'+ data_json )
        #  解析json数据
        #data_list = json.loads(data_json)
        obj = data_json.split(',')
        if '1' in msg.value.decode('utf8'):
            inl = obj[1]
        elif '2' in msg.value.decode('utf8'):
            outl = obj[1]
        else:
            continue
        result = str(inl) + ',' + str(outl)
        print(result)
        socketio.emit('test_message2',{'data':result})
#***********************************************************************

@socketio.on('test_connect')
def connect(message):
    print(message)
    global thread
    if thread is None:
        thread = socketio.start_background_task(target=background_thread2)#**************************
    socketio.emit('connected', {'data': 'Connected'})


@app.route("/")
def handle_mes():
    return render_template("index.html")


if __name__ == '__main__':
    socketio.run(app,debug=True)